package main

// go get -u github.com/eclipse/paho.mqtt.golang
import (
	"bufio"
	"encoding/json"
	"fmt"
	"os"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)

var client mqtt.Client
var (
	url      = "tcp://139.129.198.23:1883"
	clientID = "go_mqtt_client1"
	userName = "mqtt_xinjiang"
	passWord = "eDSsHdj9RP"
	topic    = "deviceInfo/1097858058140254208"
)

// 初始化MQTT服务
func NewClient() {
	if client == nil {
		opts := mqtt.NewClientOptions()
		opts.AddBroker(url) // 这个中转服务器不需要任何账号密码
		opts.SetClientID(clientID)
		opts.SetUsername(userName)
		opts.SetPassword(passWord)

		opts.OnConnect = func(c mqtt.Client) {
			fmt.Println("MQTT链接成功!")
		}
		opts.OnConnectionLost = func(c mqtt.Client, err error) {
			fmt.Println("MQTT断开链接!", err.Error())
			fmt.Println("尝试重新链接!")
			NewClient()
		}
		client = mqtt.NewClient(opts)
	}

	if token := client.Connect(); token.Wait() && token.Error() != nil {
		fmt.Println(token.Error())
	}

	for {
		// 订阅事件
		for _, subscribe := range subscribes {
			subscribe()
		}
	}
}

// 发布消息 ClientSend("topic/publish/example", 2, false, `{"code":0, "msg":"hello world!"}`)
func ClientSend(topic string, qos byte, retained bool, payload interface{}) error {
	if token := client.Publish(topic, qos, retained, payload); token.Wait() && token.Error() != nil {
		fmt.Println("消息发布失败！", token.Error())
		return token.Error()
	}
	return nil
}

// 订阅消息
func ClientSubscribe(topic string, qos byte, callback mqtt.MessageHandler, err func(error)) {
	if token := client.Subscribe(topic, qos, func(c mqtt.Client, msg mqtt.Message) {
		callback(c, msg)
	}); token.Wait() && token.Error() != nil {
		err(token.Error())
	}
}

// 订阅消息
var subscribes = []func(){

	// 调用
	// subscribeExample,
	subscribeIOT,
}

func subscribeExample() {
	ClientSubscribe(topic, 1, func(c mqtt.Client, msg mqtt.Message) {
		fmt.Println("subscribe Msg:", string(msg.Payload()))
	}, func(err error) {
		fmt.Println(err.Error())
	})
}

// /----------------------------------------------------
func subscribeIOT() {
	ClientSubscribe(topic, 1, func(c mqtt.Client, msg mqtt.Message) {
		var result map[string]string
		json.Unmarshal(msg.Payload(), &result)
		key := result["deviceID"]
		if _, ok := summarize[key]; ok { // 消息是否是map中的key,  如果是结束数量+1
			summarize[key] = summarize[key] + 1
			a := time.Now().Format(time.DateTime) + string(msg.Payload())
			fmt.Println(a)

			fmt.Println("")
			fmt.Println(summarize)
			fmt.Println("")
			go wfile(a, summarize)
		}
	}, func(err error) {
		fmt.Println(err.Error())
	})
}

var summarize = map[string]int{
	"1105923130930495488": 0,
	"1105928399093760000": 0,
	"1105929022992285696": 0,
	"1105929379189358592": 0,
	"1105929952580075520": 0,
	"1105930418542084096": 0,
	"1105930972223766528": 0,
	"1105931963052589056": 0,
	"1105932358537707520": 0,
	"1105932778328817664": 0,
}

func wfile(msg string, summarize map[string]int) {
	file, _ := os.OpenFile("receive.txt", os.O_CREATE|os.O_RDWR|os.O_APPEND, 0644)
	defer file.Close()
	wr := bufio.NewWriter(file)

	wr.WriteString(msg+"\n")
	wr.WriteString("\n")

	data, _ := json.Marshal(summarize)
	wr.WriteString(string(data) + "\n")
	wr.WriteString("------------------------------------\n")
	wr.Flush()

}
